package com.meetup.memcached;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class Replication {
//	private static Replication replication = new Replication();
	private Integer repliNum; //副本数
	private ExecutorService exec = Executors.newFixedThreadPool(200);
	private Map<String, Long> tmpMap = new HashMap<String, Long>();
	private StringBuilder oldHostlist = new StringBuilder();
	
	// pool instance
	private SockIOPool pool;
	
	public Replication(Integer repliNum, SockIOPool pool) {
		this.repliNum = repliNum;
		this.pool = pool;
	}
	
//	public static Replication getReplicationInstance() {
//		return replication;
//	}
	
	/**
	 * 开启多线程并行执行set操作和复制操作
	 * @param pool
	 * @param repliNum  副本数
	 */
	public boolean operSetRepliProcess(String key, Object value, Integer hashCode, MemcachedClient mClient) {
		SetTask setTask = null;
		Boolean bool = null;
		List<Future<Boolean>> lst_future = new ArrayList<Future<Boolean>>();
		long bucket = pool.getBucket(key, hashCode);
		Long newBucket = new Long(bucket);
		for (int i=1 ; i<=repliNum ; i++) {
			System.out.println("i=="+i);
			if (i == 1) {
				setTask = new SetTask(key, value, bucket, mClient);
			} else {
				newBucket = getClockwiseVNodeHash(newBucket);
				setTask = new SetTask(key, value, newBucket, mClient);
			}
			Future<Boolean> future = exec.submit(setTask);
			lst_future.add(future);
		}
		
		for (Future<Boolean> future : lst_future) {
			try {
				try {
					bool = future.get(20, TimeUnit.SECONDS);
				} catch (TimeoutException e) {
					e.printStackTrace();
					throw new RuntimeException("set操作复制节点失败: 操作超时!");
				}
				System.out.println("bool=="+bool);
				if (!bool) {
					throw new RuntimeException("set操作复制节点失败");
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		return bool;
	}
	
	/**
	 * 顺时针方向获取下一个物理节点的虚拟节点hash
	 * @param hv
	 * @return
	 */
	private Long getClockwiseVNodeHash(Long hv) {
		String host = null;
		Long newHv = null;
		SortedMap<Long,String> tmap = pool.getConsistentBuckets().tailMap(hv);
		Iterator<Map.Entry<Long,String>> iterator = tmap.entrySet().iterator();
		while(iterator.hasNext()){
			Map.Entry<Long,String> me = iterator.next();
			if (me.getKey().equals(hv)) {
				host = me.getValue();
				if (oldHostlist.length()==0) {
					oldHostlist.append(me.getValue());
					oldHostlist.append(";");
				}
				System.out.println("old host=="+me.getValue());
				continue;
			}
			if (!me.getValue().equals(host)) {
				if (oldHostlist.indexOf(me.getValue()) != -1) {
					continue;
				} 
				newHv = me.getKey();
				oldHostlist.append(me.getValue());
				oldHostlist.append(";");
				System.out.println("new host=="+me.getValue());
				break;
			}
		}
		System.out.println("oldHostlist=="+oldHostlist);
		if (newHv == null) {
			throw new RuntimeException("满足复制副本数的节点不够: 副本数="+repliNum+";");
		}
		return newHv;
	}
	
	private class SetTask implements Callable<Boolean> {
		private MemcachedClient mClient;
		private long bucket;
		private String key;
		private Object value;
		
		public SetTask(String key, Object value, long bucket, MemcachedClient mClient) {
			this.key = key;
			this.value = value;
			this.bucket = bucket;
			this.mClient = mClient;
		}
		
		@Override
		public Boolean call() throws Exception {
			return this.mClient.set("set", key, value, null, bucket, false);
		}
		
	}
}
